/**
 * Copyright (c) iwindplus Technologies Co., Ltd.2024-2030, All rights reserved.
 */

package com.iwindplus.boot.rocket.service.impl;

import cn.hutool.core.text.CharSequenceUtil;
import com.iwindplus.boot.domain.dto.MessageDTO;
import com.iwindplus.boot.rocket.domain.dto.RocketMessageDTO;
import com.iwindplus.boot.rocket.service.RocketService;
import jakarta.annotation.Resource;
import java.util.Optional;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;

/**
 * rocket业务层接口实现类.
 *
 * @param <T> 泛型
 * @author zengdegui
 * @since 2023/10/15 11:29
 */
@Slf4j
public class RocketServiceImpl<T> implements RocketService<T> {

    @Resource
    private RocketMQTemplate rocketTemplate;

    @Override
    public RocketMQTemplate getRocketTemplate() {
        return this.rocketTemplate;
    }

    @Override
    public SendResult syncSend(RocketMessageDTO<T> entity) {
        log.info("rocketmq 同步消息发送 = {}", entity);
        final DefaultMQProducer producer = this.rocketTemplate.getProducer();
        final Message<MessageDTO<T>> message = MessageBuilder.withPayload(entity.getMessage()).build();
        SendResult sendResult;
        if (CharSequenceUtil.isNotBlank(entity.getHashKey())) {
            sendResult = this.rocketTemplate.syncSendOrderly(entity.getTopic(), message, entity.getHashKey(),
                producer.getSendMsgTimeout(), Optional.ofNullable(entity.getDelayLevel()).orElse(0));
        } else {
            sendResult = this.rocketTemplate.syncSend(entity.getTopic(), message,
                producer.getSendMsgTimeout(), Optional.ofNullable(entity.getDelayLevel()).orElse(0));
        }
        return sendResult;
    }

    @Override
    public void asyncSend(RocketMessageDTO<T> entity) {
        log.info("rocketmq 异步消息发送 = {}", entity);
        final DefaultMQProducer producer = this.rocketTemplate.getProducer();
        final Message<MessageDTO<T>> message = MessageBuilder.withPayload(entity.getMessage()).build();
        if (CharSequenceUtil.isNotBlank(entity.getHashKey())) {
            this.rocketTemplate.asyncSendOrderly(entity.getTopic(), message, entity.getHashKey(),
                asyncSendCallback(entity.getMessage().getTaskId(), entity.getTopic()),
                producer.getSendMsgTimeout(), Optional.ofNullable(entity.getDelayLevel()).orElse(0));
        } else {
            this.rocketTemplate.asyncSend(entity.getTopic(), message,
                asyncSendCallback(entity.getMessage().getTaskId(), entity.getTopic()),
                producer.getSendMsgTimeout(), Optional.ofNullable(entity.getDelayLevel()).orElse(0));
        }
    }

    @Override
    public void sendOneWay(RocketMessageDTO<T> entity) {
        log.info("rocketmq 单向消息发送 = {}", entity);
        if (CharSequenceUtil.isNotBlank(entity.getHashKey())) {
            this.rocketTemplate.sendOneWayOrderly(entity.getTopic(), entity.getMessage(), entity.getHashKey());
        } else {
            this.rocketTemplate.sendOneWay(entity.getTopic(), entity.getMessage());
        }
    }

    private static SendCallback asyncSendCallback(String taskId, String topic) {
        return new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("rocketmq 异步消息发送成功 = [TaskId={}] - [Topic={}] - [SendStatus={}]", taskId, topic, sendResult.getSendStatus());
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("rocketmq 异步消息发送失败 = [TaskId={}] - [Topic={}] - [ErrorMessage={}]", taskId, topic, throwable.getMessage());
            }
        };
    }
}
